package com.neusoft.databus.agent.service.impl;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.atomic.AtomicLong;
import java.util.stream.Collectors;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;

import org.apache.flume.Event;
import org.apache.flume.EventDeliveryException;
import org.apache.flume.FlumeException;
import org.apache.flume.agent.embedded.EmbeddedAgent;
import org.apache.flume.event.EventBuilder;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import com.google.common.base.Charsets;
import com.neusoft.bizcore.web.utils.JsonUtils;
import com.neusoft.databus.agent.config.DatabusAgentProperties;
import com.neusoft.databus.common.bean.CommandBean;
import com.neusoft.databus.common.constant.DatabusErrorCode;

import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;

@Slf4j
@Service
public class CommandFlumeAgentService {

    @Autowired
    private DatabusAgentProperties properties;

    private EmbeddedAgent embeddedAgent = null;
    /** Agent状态 */
    @Getter
    @Setter
    private boolean started = false;

    private ArrayBlockingQueue<CommandBean> queue;

    private Consumer consumer;

    private final AtomicLong total = new AtomicLong(0);

    @PostConstruct
    public void init() {
        this.embeddedAgent = new EmbeddedAgent("embededAgent");
        this.start();
        this.queue = new ArrayBlockingQueue<>(100000);
        this.consumer = new Consumer(this.embeddedAgent, this.queue);
        new Thread(this.consumer).start();
    }

    @PreDestroy
    public void dostory() {
        try {
            this.embeddedAgent.stop();
        } catch (final FlumeException e) {
            if (CommandFlumeAgentService.log.isErrorEnabled()) {
                CommandFlumeAgentService.log.error(e.getMessage());
            }
        }
    }

    public void start() {
        final Map<String, String> config = new HashMap<>();

        config.put("channel.type", "memory");
        config.put("channel.capacity", this.properties.getFlumeChannel().getCapacity());
        config.put("channel.transactionCapacity", this.properties.getFlumeChannel().getTransactionCapacity());
        config.put("channel.keep-alive", this.properties.getFlumeChannel().getKeepAlive());

        //原始数据启动n个sink从channel中读取数据
        final String rpcServerStr = this.properties.getFlumeSink().getAvroRpcServer();
        final String[] rpcServers = rpcServerStr.split(",");

        final int sinkCount = rpcServers.length;
        String sinks = "";
        for (int i = 0; i < sinkCount; i++) {
            sinks += "avroSk" + i + (i == (sinkCount - 1) ? "" : " ");
        }
        config.put("sinks", sinks);

        for (int i = 0; i < sinkCount; i++) {
            config.put("avroSk" + i + ".type", "avro");

            final String[] hostPortPair = rpcServers[i].split(":");
            config.put("avroSk" + i + ".hostname", hostPortPair[0]);
            config.put("avroSk" + i + ".port", hostPortPair[1]);
            config.put("avroSk" + i + ".batch-size", this.properties.getFlumeSink().getBatchSize());
            config.put("avroSk" + i + ".connect-timeout", this.properties.getFlumeSink().getConnectionTimeout());
            config.put("avroSk" + i + ".request-timeout", this.properties.getFlumeSink().getRequestTimeout());
        }
        config.put("processor.type", this.properties.getFlumeProcessorType());

        try {
            this.embeddedAgent.configure(config);
            this.embeddedAgent.start();
            this.started = true;
        } catch (final FlumeException e) {
            if (CommandFlumeAgentService.log.isErrorEnabled()) {
                CommandFlumeAgentService.log.error(e.getMessage());
            }
        }
    }

    public void stop() {
        try {
            this.embeddedAgent.stop();
            this.started = false;
        } catch (final FlumeException e) {
            if (CommandFlumeAgentService.log.isErrorEnabled()) {
                CommandFlumeAgentService.log.error(e.getMessage());
            }
        }
    }

    /**
     * 发送
     *
     * @param messageBeans
     */
    public void send(final CommandBean bean) {
        List<CommandBean> subList = new ArrayList<>();
        if (bean.isError()) {
            subList.add(bean);
        } else {
            subList = bean.getResourceBean().stream().map(it -> {
                final CommandBean b = new CommandBean();
                b.setBeginTime(bean.getBeginTime());
                b.setAsync(bean.isAsync());
                b.setCollectBeginTime(bean.getCollectBeginTime());
                b.setCollectEndTime(bean.getCollectEndTime());
                b.setConnection(bean.getConnection());
                b.setConnectionStr(bean.getConnectionStr());
                b.setConnectRetry(bean.getConnectRetry());
                b.setConnectTimeout(bean.getConnectTimeout());
                b.setConnType(bean.getConnType());
                b.setEndTime(bean.getEndTime());
                b.setErrorCode(bean.getErrorCode());
                b.setForTest(bean.isForTest());
                b.setGateway(bean.getGateway());
                b.setProtocol(bean.getProtocol());
                b.setProtocolConfig(bean.getProtocolConfig());
                b.getResourceBean().add(it);
                b.setUsageType(bean.getUsageType());

                if (b.getResourceBean().get(0).getMetrics().stream().filter(m -> m.isParseSuccess()).count() == 0) {
                    b.setErrorCode(DatabusErrorCode.ERR_CODE_406004);
                }
                return b;
            }).collect(Collectors.toList());
        }

        this.total.addAndGet(1);
        if (CommandFlumeAgentService.log.isDebugEnabled()) {
            CommandFlumeAgentService.log.debug("flume receive {} events.", subList.size());
        }
        try {
            for (final CommandBean b : subList) {
                this.queue.put(b);
            }
        } catch (final InterruptedException e) {
            this.send(bean);
            if (CommandFlumeAgentService.log.isErrorEnabled()) {
                CommandFlumeAgentService.log.error(e.getMessage());
            }
        }

    }

    @Slf4j
    static class Consumer implements Runnable {
        long lastTime = System.currentTimeMillis();
        ArrayBlockingQueue<CommandBean> queue;
        EmbeddedAgent embeddedAgent;

        public Consumer(final EmbeddedAgent embeddedAgent, final ArrayBlockingQueue<CommandBean> queue) {
            this.embeddedAgent = embeddedAgent;
            this.queue = queue;
        }

        @Override
        public void run() {
            while (true) {
                if ((this.queue.size() >= 10) || ((System.currentTimeMillis() - this.lastTime) > 1000)) {
                    this.lastTime = System.currentTimeMillis();

                    final List<CommandBean> messageBeans = new ArrayList<>();
                    while (!this.queue.isEmpty()) {
                        messageBeans.add(this.queue.poll());
                    }

                    if (messageBeans.size() == 0) {
                        continue;
                    }

                    final List<Event> list = new ArrayList<>();
                    for (final CommandBean command : messageBeans) {
                        final String jsonStr = JsonUtils.pojoToJson(command);
                        final Event event = EventBuilder.withBody(jsonStr.getBytes(Charsets.UTF_8));
                        final String protocol = command.getProtocol().getValue();
                        if ("PLUGIN".equals(protocol)) {
                            final String config = command.getProtocolConfig();
                            final Map<String, Object> configMap = JsonUtils.jsonToMap(config);
                            final String parserName = (String) configMap.get("protocolName");
                            event.getHeaders().put("type", parserName.toUpperCase());
                        } else {
                            event.getHeaders().put("type", command.getProtocol().getValue());
                        }
                        list.add(event);
                    }

                    try {
                        if (Consumer.log.isDebugEnabled()) {
                            Consumer.log.debug("send {} events by flume", list.size());
                            final Map<String, Integer> statis = new HashMap<>();
                            list.stream().map(it -> it.getHeaders().get("type")).collect(Collectors.toList())
                                    .forEach(type -> {
                                        statis.put(type, (statis.containsKey(type) ? statis.get(type) : 0) + 1);
                                    });
                            final StringBuilder sb = new StringBuilder();
                            for (final Entry<String, Integer> e : statis.entrySet()) {
                                sb.append(e.getKey()).append(":").append(e.getValue()).append(" ");
                            }
                            Consumer.log.debug(sb.toString());

                        }
                        this.embeddedAgent.putAll(list);
                    } catch (final EventDeliveryException e) {
                        if (Consumer.log.isErrorEnabled()) {
                            Consumer.log.error(e.getMessage());
                        }
                    }
                } else {
                    try {
                        Thread.sleep(100);
                    } catch (final InterruptedException e) {
                        if (Consumer.log.isErrorEnabled()) {
                            Consumer.log.error(e.getMessage());
                        }
                    }
                }

            }
        }
    }

}
